package com.wdl.networkflow.uv

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

import java.lang

/**
 * 自定义 ProcessWindowFunction，把当前数据进行处理，位图保存在 redis 中
 */
class UvCountResultWithBloomFilter extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow]{

  var jedis: Jedis = _
  var bloom: UvCountBloomFilter = _
  override def open(parameters: Configuration): Unit = {
    jedis = new Jedis("", 6379)
    /** 位图大小 10亿个位， 也就是2^30，占用 128MB */
    bloom = new UvCountBloomFilter(1 << 30)
  }

  /** 每来一个数据，主要是要用布隆过滤器判断 redis 位图中对应位置是否为 1 */
  override def process(key: String,
                       context: Context,
                       elements: Iterable[(String, Long)],
                       out: Collector[UvCount]): Unit = {
    /** bitMap 用当前窗口的 end 作为 key，保存到 redis 里，(windowEnd, bitMap) */
    val storeKey: String = context.window.getEnd.toString

    /** 把每个窗口的 uv count 值，作为状态也存入 redis 中，存成一张叫做 countMap 的表 */
    val countMap = "countMap"
    var count = 0L
    if (jedis.hget(countMap, storeKey) != null) {
      count = jedis.hget(countMap, storeKey).toLong

      /** 取 userId，计算 hash 值，判断是否在位图中 */
      val userId = elements.last._2.toString
      val offset = bloom.hash(userId, 61)
      val isExist = jedis.getbit(storeKey, offset)
      /** 如果不存在就将对应位置设置为 1，如果存在，不需要操作 */
      if ( !isExist ){
        jedis.setbit(storeKey, offset, true)
        jedis.hset(countMap, storeKey, (count + 1).toString)
      }
    }

  }
}
